订单明细表：
select
    data['id'] AS id,
    data['order_id'] AS order_id,
    data['sku_id'] AS sku_id,
    data['sku_name'] AS sku_name,
    data['order_price'] AS order_price,
    data['sku_num'] AS sku_num,
    data['create_time'] AS create_time,
    data['source_type'] AS source_type,
    data['source_id'] AS source_id,
    data['split_total_amount'] AS split_total_amount,
    data['split_activity_amount'] AS split_activity_amount,
    data['split_coupon_amount'] AS split_coupon_amount,
    pt
from topic_db
where `database` = 'gmall'
and `table` = 'order_detail'


insert into order_detail(order_id,sku_id,sku_name,img_url,order_price,sku_num,source_type) values(5093,33,'aaaa','aaaaaa',488,3,2403);


select
    data['id'] AS id,
    data['consignee'] AS consignee,
    data['consignee_tel'] AS consignee_tel,
    data['total_amount'] AS total_amount,
    data['order_status'] AS order_status,
    data['user_id'] AS user_id,
    data['payment_way'] AS payment_way,
    data['delivery_address'] AS delivery_address,
    data['order_comment'] AS order_comment,
    data['out_trade_no'] AS out_trade_no,
    data['trade_body'] AS trade_body,
    data['create_time'] AS create_time,
    data['operate_time'] AS operate_time,
    data['expire_time'] AS expire_time,
    data['process_status'] AS process_status,
    data['tracking_no'] AS tracking_no,
    data['parent_order_id'] AS parent_order_id,
    data['province_id'] AS province_id,
    data['activity_reduce_amount'] AS activity_reduce_amount,
    data['coupon_reduce_amount'] AS coupon_reduce_amount,
    data['original_total_amount'] AS original_total_amount,
    data['feight_fee'] AS feight_fee,
    data['feight_fee_reduce'] AS feight_fee_reduce,
    data[']refundable_time'] AS refundable_time,
    `type`,
    `old`
from topic_db
where `database` = 'gmall'
and `table` = 'order_info'

insert into order_info(consignee,consignee_tel,total_amount,order_status,user_id,delivery_address,order_comment,out_trade_no) values
('张三','12712221112',5968,'1004',4038,'aaa','aaaa','1233333333333333');

update order_info set consignee='张三san' where consignee='张三';


只有新增
select
    data['id'] AS id,
    data['order_id'] AS order_id,
    data['order_detail_id'] AS order_detail_id,
    data['activity_id'] AS activity_id,
    data['activity_rule_id'] AS activity_rule_id,
    data['sku_id'] AS sku_id,
    data['create_time'] AS create_time,
from topic_db
where `database` = 'gmall'
and `table` = 'order_detail_activity'

insert into order_detail_activity(order_id,order_detail_id,activity_id,activity_rule_id,sku_id) values(5092,13610,2,4,12);

[只有新增]
select
    data['id'] AS id,
    data['order_id'] AS order_id,
    data['order_detail_id'] AS order_detail_id,
    data['coupon_id'] AS coupon_id,
    data['coupon_use_id'] AS coupon_use_id,
    data['sku_id'] AS sku_id,
    data['create_time'] AS create_time
from topic_db
where `database` = 'gmall'
and `table` = 'order_detail_coupon'

insert into order_detail_coupon(order_id,order_detail_id,coupon_id,coupon_use_id,sku_id) values(5092,13610,2,4,12);


[JOIN]

select
    od.id,
    od.order_id,
    od.sku_id,
    od.sku_name,
    od.order_price,
    od.sku_num,
    od.create_time,
    od.source_type,
    od.source_id,
    od.split_total_amount,
    od.split_activity_amount,
    od.split_coupon_amount,
    oi.consignee,
    oi.consignee_tel,
    oi.total_amount,
    oi.order_status,
    oi.user_id,
    oi.payment_way,
    oi.delivery_address,
    oi.order_comment,
    oi.out_trade_no,
    oi.trade_body,
    oi.create_time,
    oi.operate_time,
    oi.expire_time,
    oi.process_status,
    oi.tracking_no,
    oi.parent_order_id,
    oi.province_id,
    oi.activity_reduce_amount,
    oi.coupon_reduce_amount,
    oi.original_total_amount,
    oi.feight_fee,
    oi.feight_fee_reduce,
    oi.refundable_time,
    oa.id as order_detail_activity_id,
    oa.activity_id,
    oa.activity_rule_id,
    oc.id as order_detail_coupon_id,
    oc.coupon_id,
    oc.coupon_use_id,
    oi.`type`,
    oi.`old`
from order_detail_table od
join order_info_table oi on od.order_id=oi.id
left join order_detail_table oa on od.id=oa.order_detail_id
left join order_coupon_table oc on od.id=oc.order_detail_id
JOIN base_dic FOR SYSTEM_TIME AS OF od.pt as dic on od.source_type=dic.dic_code


create table dwd_order_pre(
   `id`  string,
   `order_id`  string,
   `sku_id`  string,
   `sku_name`  string,
   `order_price`  string,
   `sku_num`  string,
   `create_time`  string,
   `source_type_id`  string,
   `source_type_name`  string,
   `source_id`  string,
   `split_total_amount`  string,
   `split_activity_amount`  string,
   `split_coupon_amount`  string,
   `consignee`  string,
   `consignee_tel`  string,
   `total_amount`  string,
   `order_status`  string,
   `user_id`  string,
   `payment_way`  string,
   `delivery_address`  string,
   `order_comment`  string,
   `out_trade_no`  string,
   `trade_body`  string,
   `create_time`  string,
   `operate_time`  string,
   `expire_time`  string,
   `process_status`  string,
   `tracking_no`  string,
   `parent_order_id`  string,
   `province_id`  string,
   `activity_reduce_amount`  string,
   `coupon_reduce_amount`  string,
   `original_total_amount`  string,
   `feight_fee`  string,
   `feight_fee_reduce`  string,
   `refundable_time`  string,
   `order_detail_activity_id`  string,
   `activity_id`  string,
   `activity_rule_id`  string,
   `order_detail_coupon_id`  string,
   `coupon_id`  string,
   `coupon_use_id`  string,
   `type`  string,
   `old`   map<string,string>
)

*********************************DWS层**************************************

create table page_log(
  `page` map<string,string>,
  `ts`  bigint,
  `rt`  TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)),  -- TIMESTAMP(3),
    -- declare user_action_time as event time attribute and use 5 seconds delayed watermark strategy
    WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND

)

select
    page['item'] item,
    rt
from page_log
where page['last_page_id'] = 'search'
and page['item_type'] = 'keyword'
and page['item'] is not null

SELECT
    word,
    rt
FROM filterTable, LATERAL TABLE(SplitFunction(item))

select
    DATE_FORMAT(TUMBLE_START(rt, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt,
    DATE_FORMAT(TUMBLE_END(rt, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt,
    'search' source,
    word,
    count(*) ct
from splitTable
group by word,TUMBLE(rt, INTERVAL '10' SECOND)

***************clickhouse******
use gmall;
drop table if exists dws_traffic_source_keyword_page_view_window;
create table if not exists dws_traffic_source_keyword_page_view_window
(
    stt           DateTime,
    edt           DateTime,
    source        String,
    keyword       String,
    keyword_count UInt64,
    ts            UInt64
) engine = ReplacingMergeTree(ts)
      partition by toYYYYMMDD(stt)
      order by (stt, edt, source, keyword);

drop table if exists dws_traffic_vc_ch_ar_is_new_page_view_window;
create table if not exists dws_traffic_vc_ch_ar_is_new_page_view_window
(
    stt     DateTime,
    edt     DateTime,
    vc      String,
    ch      String,
    ar      String,
    is_new  String,
    uv_ct   UInt64,
    sv_ct   UInt64,
    pv_ct   UInt64,
    dur_sum UInt64,
    uj_ct   UInt64,
    ts      UInt64
) engine = ReplacingMergeTree(ts)
      partition by toYYYYMMDD(stt)
      order by (stt, edt, vc, ch, ar, is_new);


  drop table if exists dws_traffic_page_view_window;
  create table if not exists dws_traffic_page_view_window
  (
      stt               DateTime,
      edt               DateTime,
      home_uv_ct        UInt64,
      good_detail_uv_ct UInt64,
      ts                UInt64
  ) engine = ReplacingMergeTree(ts)
        partition by toYYYYMMDD(stt)
        order by (stt, edt);

select * from dws_traffic_page_view_window


drop table if exists dws_traffic_page_view_window;
create table if not exists dws_traffic_page_view_window
(
    stt               DateTime,
    edt               DateTime,
    home_uv_ct        UInt64,
    good_detail_uv_ct UInt64,
    ts                UInt64
) engine = ReplacingMergeTree(ts)
      partition by toYYYYMMDD(stt)
      order by (stt, edt);